package com.lsx143.wordcount.util;

import com.lsx143.wordcount.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;

import java.lang.reflect.Field;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

public class MyUtil {
    private static final String SCHEME_TYPE_STRING = "java.lang.String";
    private static final String SCHEME_TYPE_LONG = "java.lang.Long";
    private static final String SCHEME_TYPE_INTEGER = "java.lang.Integer";

    public static <T> List<T> toList(Iterable<T> it) {
        List<T> list = new ArrayList<>();
        for (T t : it) {
            list.add(t);
        }
        return list;
    }

    public static String toTime(long time) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        return simpleDateFormat.format(new Date(time));
    }

    public static void main(String[] args) {
        getSchemaByBean(WaterSensor.class);
    }

    public static Schema getSchemaByBean(Class clazz) {
        Field[] fields = clazz.getDeclaredFields();
        List<String> strFields = new ArrayList<>();
        List<String> strTypes = new ArrayList<>();
        for (Field field : fields) {
            strFields.add(field.getName());
            strTypes.add(field.getGenericType().getTypeName());
        }
        Schema schema = new Schema();
        for (int i = 0; i < strFields.size(); i++) {
            switch (strTypes.get(i)) {
                case SCHEME_TYPE_STRING:
                    schema.field(strFields.get(i), DataTypes.STRING());
                    break;
                case SCHEME_TYPE_LONG:
                    schema.field(strFields.get(i), DataTypes.BIGINT());
                    break;
                case SCHEME_TYPE_INTEGER:
                    schema.field(strFields.get(i), DataTypes.INT());
                    break;
                default:
                    throw new IllegalStateException("Unexpected value: " + strTypes.get(i));
            }
        }
        return schema;
    }

    public static StreamTableEnvironment getTableEnvOnFile(StreamExecutionEnvironment env, String file) {
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Schema schema = MyUtil.getSchemaByBean(WaterSensor.class);
        tableEnv.connect(new FileSystem().path(file))
                .withFormat(new Csv().fieldDelimiter(',').lineDelimiter("\n"))
                .withSchema(schema)
                .createTemporaryTable("sensor");
        return tableEnv;
    }
}
